package com.spx.chapter09;

import com.spx.chapter05.pojo.Event;
import com.spx.util.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * create by undeRdoG on  2022-05-05  21:48
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class PeriodicPvExampleByValueState {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );


        stream.print("stream");

        // 统计每个用户的PV
        stream.keyBy(data -> data.user)
                .process(new PeriodicPvFunction())
                .print("pv");


        env.execute();


    }


    public static class PeriodicPvFunction extends KeyedProcessFunction<String,Event,String> {

        private ValueState<Long> state;

        private ValueState<Long> hasTimerState;

        @Override
        public void open(Configuration parameters) throws Exception {
            this.state = getRuntimeContext().getState(new ValueStateDescriptor<Long>("countState", Long.class));
            this.hasTimerState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("hasTimerState", Long.class));
        }

        @Override
        public void processElement(Event value, Context ctx, Collector<String> out) throws Exception {
            // 状态值+1
            Long currentCount = this.state.value();
            if (currentCount == null){
                this.state.update(1L);
            }else{
                this.state.update(currentCount + 1L);
            }

            // 注册定时器：如果当前没有注册定时器，那么就注册，否则就不注册，根据hasTimerState来判断

            Long hasTimer = hasTimerState.value();
            if (hasTimer == null){
                // 没有定时器，需要去注册
                ctx.timerService().registerEventTimeTimer(value.timestamp + 10 * 1000L);

                // 标记状态，代表当前已经注册了定时器
                this.hasTimerState.update(value.timestamp + 10 * 1000L);
            }

        }


        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

            out.collect(ctx.getCurrentKey() + " pv is " + this.state.value());

            // 清空定时器
            this.hasTimerState.clear();
        }
    }

}
